public void jsonSerializationOfExecutorRegistration() throws IOException {

    ObjectMapper mapper = new ObjectMapper();

    AppExecId appId = new AppExecId("foo", "bar");

    String appIdJson = mapper.writeValueAsString(appId);

    AppExecId parsedAppId = mapper.readValue(appIdJson, AppExecId.class);

    assertEquals(parsedAppId, appId);



    ExecutorShuffleInfo shuffleInfo =

      new ExecutorShuffleInfo(new String[]{"/bippy", "/flippy"}, 7, "hash");

    String shuffleJson = mapper.writeValueAsString(shuffleInfo);

    ExecutorShuffleInfo parsedShuffleInfo =

      mapper.readValue(shuffleJson, ExecutorShuffleInfo.class);

    assertEquals(parsedShuffleInfo, shuffleInfo);



    String legacyAppIdJson = "{"appId":"foo", "execId":"bar"}";

    assertEquals(appId, mapper.readValue(legacyAppIdJson, AppExecId.class));

    String legacyShuffleJson = "{"localDirs": ["/bippy", "/flippy"], " +

      ""subDirsPerLocalDir": 7, "shuffleManager": "hash"}";

    assertEquals(shuffleInfo, mapper.readValue(legacyShuffleJson, ExecutorShuffleInfo.class));

  }

  public void noCleanupAndCleanup() throws IOException {

    TestShuffleDataContext dataContext = createSomeData();



    ExternalShuffleBlockResolver resolver =

      new ExternalShuffleBlockResolver(conf, null, sameThreadExecutor);

    resolver.registerExecutor("app", "exec0", dataContext.createExecutorInfo("shuffleMgr"));

    resolver.applicationRemoved("app", false /* cleanup */);



    assertStillThere(dataContext);



    resolver.registerExecutor("app", "exec1", dataContext.createExecutorInfo("shuffleMgr"));

    resolver.applicationRemoved("app", true /* cleanup */);



    assertCleanedUp(dataContext);

  }